package cn.tedu.charging.device.canal;

import cn.tedu.charging.common.pojo.po.ChargingStationPO;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 充电站数据 Canal 监听服务
 * 使用阿里巴巴官方 canal.client 实现数据同步
 */
@Slf4j
@Service
public class StationCanalService {

    @Autowired
    private CanalConnector canalConnector;

    private ExecutorService executorService;

    private volatile boolean running = false;

    @PostConstruct
    public void start() {
        executorService = Executors.newSingleThreadExecutor();
        running = true;
        executorService.submit(new CanalWorker());
        log.info("充电站 Canal 监听服务已启动");
    }

    @PreDestroy
    public void stop() {
        running = false;
        if (executorService != null) {
            executorService.shutdown();
        }
        log.info("充电站 Canal 监听服务已停止");
    }

    private class CanalWorker implements Runnable {
        @Override
        public void run() {
            int retryCount = 0;
            final int maxRetries = 5;
            final long retryDelay = 5000; // 5秒
            
            while (running) {
                try {
                    // 连接 Canal Server
                    log.info("尝试连接Canal服务器...");
                    canalConnector.connect();
                    log.info("成功连接Canal服务器");
                    
                    // 订阅 charging_station 表
                    canalConnector.subscribe(".*\\.charging_station");
                    log.info("成功订阅charging_station表");
                    
                    // 回滚到未 ack 的位置
                    canalConnector.rollback();
                    log.info("回滚到未ack的位置");

                    retryCount = 0; // 重置重试计数
                    
                    while (running) {
                        // 获取消息，每次最多获取1000条记录
                        Message message = canalConnector.getWithoutAck(1000);
                        long batchId = message.getId();
                        int size = message.getEntries().size();

                        if (batchId == -1 || size == 0) {
                            // 没有数据时休眠2秒
                            Thread.sleep(2000);
                        } else {
                            // 处理数据
                            processEntries(message.getEntries());
                            // 提交确认
                            canalConnector.ack(batchId);
                        }
                    }
                } catch (Exception e) {
                    retryCount++;
                    log.error("Canal数据处理异常 (重试 {}/{})", retryCount, maxRetries, e);
                    
                    if (retryCount >= maxRetries) {
                        log.error("达到最大重试次数，停止尝试连接Canal服务器");
                        break;
                    }
                    
                    try {
                        // 断开连接（如果已连接）
                        try {
                            canalConnector.disconnect();
                        } catch (Exception disconnectEx) {
                            log.warn("断开Canal连接时发生异常", disconnectEx);
                        }
                        
                        // 等待后重试
                        Thread.sleep(retryDelay);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                } finally {
                    try {
                        canalConnector.disconnect();
                        log.info("断开与Canal服务器的连接");
                    } catch (Exception e) {
                        log.warn("断开Canal连接时发生异常", e);
                    }
                }
            }
        }
    }

    /**
     * 处理 Canal 条目数据
     * @param entries Canal 条目列表
     */
    private void processEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN 
                || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            try {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                CanalEntry.EventType eventType = rowChange.getEventType();

                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    switch (eventType) {
                        case INSERT:
                            handleInsert(rowData);
                            break;
                        case UPDATE:
                            handleUpdate(rowData);
                            break;
                        case DELETE:
                            handleDelete(rowData);
                            break;
                        default:
                            break;
                    }
                }
            } catch (Exception e) {
                log.error("解析Canal数据异常", e);
            }
        }
    }

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 处理插入操作
     * @param rowData 行数据
     */
    private void handleInsert(CanalEntry.RowData rowData) {
        ChargingStationPO station = convertToStation(rowData.getAfterColumnsList());
        log.debug("监听到充电站新增数据:{}", station);
        log.debug("新增数据开始同步到Redis");
        
        // 定义缓存的key
        String key = "station_detail_" + station.getId();
        // 删除旧数据，触发重新查询数据库
        redisTemplate.delete(key);
        log.debug("数据库新增的充电站数据Redis同步新增成功");
    }

    /**
     * 处理更新操作
     * @param rowData 行数据
     */
    private void handleUpdate(CanalEntry.RowData rowData) {
        ChargingStationPO before = convertToStation(rowData.getBeforeColumnsList());
        ChargingStationPO after = convertToStation(rowData.getAfterColumnsList());
        log.debug("监听到充电站更新数据:before:{},after:{}", before, after);
        log.debug("更新数据开始同步到Redis");
        
        // 定义缓存的key
        String key = "station_detail_" + before.getId();
        // 删除旧数据，触发重新查询数据库
        redisTemplate.delete(key);
        log.debug("数据库更新的充电站数据Redis同步更新成功");
    }

    /**
     * 处理删除操作
     * @param rowData 行数据
     */
    private void handleDelete(CanalEntry.RowData rowData) {
        ChargingStationPO station = convertToStation(rowData.getBeforeColumnsList());
        log.debug("监听到充电站删除数据:{}", station);
        log.debug("删除数据开始同步到Redis");
        
        // 定义缓存的key
        String key = "station_detail_" + station.getId();
        // 删除缓存数据
        redisTemplate.delete(key);
        log.debug("数据库删除的充电站数据Redis同步删除成功");
    }

    /**
     * 将 Canal 列数据转换为 ChargingStationPO 对象
     * @param columns 列数据列表
     * @return ChargingStationPO 对象
     */
    private ChargingStationPO convertToStation(List<CanalEntry.Column> columns) {
        ChargingStationPO station = new ChargingStationPO();
        for (CanalEntry.Column column : columns) {
            switch (column.getName()) {
                case "id":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setId(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "station_code":
                    station.setStationCode(column.getValue());
                    break;
                case "station_name":
                    station.setStationName(column.getValue());
                    break;
                case "device_number":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setDeviceNumber(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "ac_gun_number":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setAcGunNumber(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "dc_gun_number":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setDcGunNumber(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "ac_rate_power":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setAcRatePower(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "dc_rate_power":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setDcRatePower(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "province":
                    station.setProvince(column.getValue());
                    break;
                case "city":
                    station.setCity(column.getValue());
                    break;
                case "address":
                    station.setAddress(column.getValue());
                    break;
                case "station_lng":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setStationLng(new BigDecimal(column.getValue()));
                    }
                    break;
                case "station_lat":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setStationLat(new BigDecimal(column.getValue()));
                    }
                    break;
                case "device_power":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setDevicePower(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "station_model":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setStationModel(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "station_status":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setStationStatus(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "station_type":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setStationType(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "operator_id":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setOperatorId(Integer.valueOf(column.getValue()));
                    }
                    break;
                case "park_fee":
                    if (column.getValue() != null && !column.getValue().isEmpty()) {
                        station.setParkFee(new BigDecimal(column.getValue()));
                    }
                    break;
                default:
                    break;
            }
        }
        return station;
    }
}